package com.shujia.flink.fraud

import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.api.common.state.ValueState
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.util.Collector

object FraudDetector {
  val MIN_AMOUNT: Double = 1.00
  val MAX_AMOUNT: Double = 500.00
  val ONE_MINUTE: Long = 10 * 1000L
}

class FraudDetector extends KeyedProcessFunction[String, Event, Alert] {


  var alertState: ValueState[Alert] = _

  var flagState: ValueState[Boolean] = _
  var timerState: ValueState[Long] = _

  override def open(parameters: Configuration): Unit = {
    val valueStateDescriptor = new ValueStateDescriptor[Alert]("alert", classOf[Alert])
    //保存告诫信息
    alertState = getRuntimeContext.getState(valueStateDescriptor)

    val flagValueStateDescriptor = new ValueStateDescriptor[Boolean]("flag", classOf[Boolean], false)
    //标记告诫状态
    flagState = getRuntimeContext.getState(flagValueStateDescriptor)


    val timerValueStateDescriptor = new ValueStateDescriptor[Long]("timer", classOf[Long])
    //标记告诫状态
    timerState = getRuntimeContext.getState(timerValueStateDescriptor)


  }

  //processElement  每一个key的每一条数据都会执行一次
  override def processElement(value: Event, ctx: KeyedProcessFunction[String, Event, Alert]#Context, out: Collector[Alert]): Unit = {
    // 前一次交易小于MIN_AMOUNT，后一次交易大于MAX_AMOUNT  发起警告
    //两次交易必须在规定时间发生

    //说明之前有一次小额交易
    if (flagState.value()) {
      //判断当前是否是一次大额交易
      if (value.amount > FraudDetector.MAX_AMOUNT) {
        val alert = alertState.value()
        alert.event2 = value

        //发起警告
        out.collect(alert)

      }
      //清空状态
      flagState.clear()
      alertState.clear()
      //删除定时器
      ctx.timerService().deleteProcessingTimeTimer(timerState.value())
    }

    if (value.amount < FraudDetector.MIN_AMOUNT) {
      flagState.update(true)

      val alert = Alert(value, null)
      alertState.update(alert)

      //在交易额小于1的时候创建定时器
      val timer = System.currentTimeMillis() + FraudDetector.ONE_MINUTE
      ctx.timerService().registerProcessingTimeTimer(timer)
      timerState.update(timer)
    }
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, Event, Alert]#OnTimerContext, out: Collector[Alert]): Unit = {

    //超过10秒清空状态
    //清空状态
    flagState.clear()
    alertState.clear()

  }
}

case class Alert(var event1: Event, var event2: Event)
